/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月19日 下午2:08:04
*/
package cn.com.guage.flink.window;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @version:1.0.1
 * @Description: （对类进行功能描述）
 * @author: yangdechao
 * @date: datedate 2021年11月19日 下午2:08:04
 */
public class WindowAllTest {
	public static void main(String[] args) throws Exception {
		// linux里面执行nc -lk 9527
		// 创建流处理执行环境
		StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
		test1(streamEnv);
		test2(streamEnv);

	}

	public static void test1(StreamExecutionEnvironment env) throws Exception {
		// 加载数据源，并执行flatMap转换
		DataStream<String> ds = env.fromElements("good good study", "day day up")
				.flatMap(new FlatMapFunction<String, String>() {
					/**
					 * 
					 */
					private static final long serialVersionUID = -7464308627786783866L;

					@Override
					public void flatMap(String value, Collector<String> out) throws Exception {
						for (String word : value.split("\\W+")) {
							out.collect(word);
						}
					}
				});
		ds.print();
		
		DataStream<Tuple2<String, Integer>> ds_map = ds.map(new MapFunction<String, Tuple2<String, Integer>>() {
			/**
			 * 
			 */
			private static final long serialVersionUID = 4651228081643315456L;

			@Override
			public Tuple2<String, Integer> map(String s) throws Exception {
				return new Tuple2<String, Integer>(s, 1);
			}
		});

		// keyBy转换，按key重分区
		KeyedStream<Tuple2<String, Integer>, Tuple> ds_keyed = ds_map.keyBy(0);

		// 输出
		ds_keyed.print();

		// 执行
		env.execute("flink keyBy transformatiion");
	}

	public static void test2(StreamExecutionEnvironment env) throws Exception {
		//DataStream<String> inputDataStream = env.fromElements("good good study", "day day up");
        DataStream<String> inputDataStream = env.socketTextStream("192.168.0.128",9528);


		// 从socket文本流读取数据
		//DataStream<String> inputDataStream = env.socketTextStream("192.168.0.128", 9528);
		DataStream<String> flatmapStream = inputDataStream.flatMap(new FlatMapFunction<String, String>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = -6913029828363778052L;

			@Override
			public void flatMap(String value, Collector<String> out) throws Exception {
				for (String word : value.split("\\W+")) {
					out.collect(word);
				}
			}
		} );
		//flatmapStream.print();
		// 通过map转换，将事件流中事件的数据类型变换为(word,1)元组形式
		SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream = flatmapStream.map(new MapFunction<String, Tuple2<String, Integer>>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = -348493736657618480L;

			@Override
			public Tuple2<String, Integer> map(String value) throws Exception {
				return Tuple2.of(value, 1);
			}
		});
		//mapStream.print();
		// keyBy转换，按key重分区
		KeyedStream<Tuple2<String, Integer>, Tuple> keybyStream =  mapStream.keyBy(0);
	//	keybyStream.print();
		keybyStream.sum(1).print();
		keybyStream.timeWindow(Time.seconds(5));
		//keybyStream.countWindow(size)
		//keybyStream.window(TumblingProcessingTimeWindows.of(Time.seconds(200)));
		//keybyStream.timeWindow(Time.seconds(5));
		env.execute("flink keyBy transformatiion");

	}
}
